﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Apache.NMS;
using Apache.NMS.ActiveMQ;
using Apache.NMS.Util;
using Apache.NMS.ActiveMQ.Commands;
using System.Collections;
using System.IO;

namespace Transport.AMQ
{
    public delegate void MessageReceivedDelegate(string message);

    class SubscribeTopic : IDisposable
    {
        public static string SUBSCRIBTION_NAME;
        public static string TOPIC_NAME;

        private readonly IConnection _connection;
        private readonly ISession _session;
        private readonly IDestination _destination;
        
        private readonly IQueueBrowser _browser;
        private readonly IConnectionFactory _factory;
        private readonly ArrayList _messagequeue = new ArrayList();
        private readonly IMessageConsumer _consumer;
        private DataProvider _provider;
        private bool isDisposed = false;
        public event MessageReceivedDelegate OnMessageReceived;

        public SubscribeTopic(string SubscribtionName, string TopicName)
        {
            SUBSCRIBTION_NAME = SubscribtionName;
            TOPIC_NAME = TopicName;
            this._factory = new ConnectionFactory();
            this._connection = _factory.CreateConnection();
            this._connection.Start();
            this._session = _connection.CreateSession();
            ActiveMQTopic topic = new ActiveMQTopic(TOPIC_NAME);
            //IQueue q = _session.GetQueue("parameters");
            this._destination = _session.GetTopic(TOPIC_NAME);
            this._consumer = _session.CreateDurableConsumer(topic, "parameters", "true", false);
            this._consumer.Listener += new MessageListener(OnMessage);
           
        }

        public void OnMessage(IMessage message)
         {
            ITextMessage textMessage = message as ITextMessage;
            if (this.OnMessageReceived != null)
            {
                this.OnMessageReceived(textMessage.Text);                
            }
        }

        #region IDisposable Members

        public void Dispose()
        {
            if (!this.isDisposed)
            {
                this._consumer.Dispose();
                this._session.Dispose();
                this._connection.Dispose();
                this.isDisposed = true;
            }
        }
        #endregion
    }
}
